signals.hpp

浏览该文件的文档.

#ifndef NEFORCE_CORE_ASYNC_SIGNALS_HPP__
#define NEFORCE_CORE_ASYNC_SIGNALS_HPP__

 
#include "NeForce/core/algorithm/remove.hpp"
#include "NeForce/core/async/mutex.hpp"
#include "NeForce/core/container/vector.hpp"
#include "NeForce/core/functional/apply.hpp"
#include "NeForce/core/functional/function.hpp"
#include "NeForce/core/memory/weak_ptr.hpp"
#include "NeForce/core/reflect/any.hpp"
NEFORCE_BEGIN_NAMESPACE__


class signal_base {
public:
    virtual ~signal_base() = default;

    virtual void disconnect_all() = 0;

    virtual void block() = 0;

    virtual void unblock() = 0;

    NEFORCE_NODISCARD virtual bool is_blocked() const = 0;

    NEFORCE_NODISCARD virtual size_t slot_count() const = 0;

    virtual void emit_dynamic(const vector<reflect::meta_any>& args) = 0;

    virtual void connect_dynamic(function<void(const vector<reflect::meta_any>&)> callback) = 0;
};
 

enum class callback_result {
    keep,  
    erase, 
};
 

struct oneshot_t {
    constexpr oneshot_t() noexcept = default;
};

NEFORCE_INLINE17 constexpr oneshot_t oneshot{};

enum class nshot_t : size_t {
};
 

class connection {
private:
    shared_ptr<bool> connected_; 
 
public:
    connection() :
    connected_(_NEFORCE make_shared<bool>(true)) {}

    void disconnect() noexcept {
        if (connected_) {
            *connected_ = false;
        }
    }

    NEFORCE_NODISCARD bool connected() const noexcept { return connected_ && *connected_; }

    NEFORCE_NODISCARD shared_ptr<bool> flag() const noexcept { return connected_; }
};
 

class scope_connection {
private:
    connection conn_; 
 
public:
    scope_connection() = default;

    explicit scope_connection(connection conn) noexcept :
    conn_(_NEFORCE move(conn)) {}

    ~scope_connection() { conn_.disconnect(); }
 
    scope_connection(const scope_connection&) = delete;
    scope_connection& operator=(const scope_connection&) = delete;

    scope_connection(scope_connection&& other) noexcept :
    conn_(_NEFORCE move(other.conn_)) {}

    scope_connection& operator=(scope_connection&& other) noexcept {
        if (addressof(other) == this) {
            return *this;
        }
        conn_.disconnect();
        conn_ = _NEFORCE move(other.conn_);
        return *this;
    }

    void disconnect() noexcept { conn_.disconnect(); }

    NEFORCE_NODISCARD bool connected() const noexcept { return conn_.connected(); }

    connection release() noexcept { return _NEFORCE move(conn_); }
};
 
 
template <typename... Types>
struct signal;

template <typename... Types>
class signal_blocker {
private:
    shared_ptr<bool> blocked_flag_; 
    bool old_value_;                
    bool released_ = false;         
 
public:
    explicit signal_blocker(signal<Types...>& sig) noexcept :
    blocked_flag_(sig.block_flag()) {
        old_value_ = *blocked_flag_;
        *blocked_flag_ = true;
    }

    ~signal_blocker() {
        if (blocked_flag_ && !released_) {
            *blocked_flag_ = old_value_;
        }
    }
 
    signal_blocker(const signal_blocker&) = delete;
    signal_blocker& operator=(const signal_blocker&) = delete;

    void unblock() noexcept {
        if (blocked_flag_ && !released_) {
            *blocked_flag_ = old_value_;
            released_ = true;
        }
    }
};
 

template <typename... Types>
struct signal final : public signal_base {
private:
    struct slot_entry {
        using callback_type = function<callback_result(Types...)>; 
 
        callback_type callback;          
        shared_ptr<bool> connected_flag; 
        int priority;                    
 
        slot_entry(callback_type cb, shared_ptr<bool> flag, const int pri = 0) :
        callback(_NEFORCE move(cb)),
        connected_flag(_NEFORCE move(flag)),
        priority(pri) {}
    };
 
    vector<slot_entry> slots_{};                                       
    shared_ptr<bool> blocked_flag_{_NEFORCE make_shared<bool>(false)}; 
    mutable mutex mutex_;                                              
 
    template <typename T>
    static shared_ptr<T> lock_if_weak(const weak_ptr<T>& self) {
        return self.lock();
    }
    template <typename T>
    static const shared_ptr<T>& lock_if_weak(const shared_ptr<T>& self) noexcept {
        return self;
    }
    template <typename T>
    static T* lock_if_weak(T* self) noexcept {
        return self;
    }
 
    template <typename Func>
    auto with_lock(Func&& func) const {
        lock<mutex> lock(mutex_);
        return _NEFORCE forward<Func>(func)();
    }
 
    template <typename Func>
    auto with_lock(Func&& func) {
        lock<mutex> lock(mutex_);
        return _NEFORCE forward<Func>(func)();
    }

    template <typename Self, typename MemFn>
    auto bind(Self self, MemFn memfn) {
        return [self = _NEFORCE move(self), memfn](Types... args) {
            auto ptr = signal::lock_if_weak(self);
            if (ptr == nullptr) {
                return callback_result::erase;
            }
            ((*ptr).*memfn)(_NEFORCE forward<Types>(args)...);
            return callback_result::keep;
        };
    }

    template <typename Self, typename MemFn>
    auto bind(Self self, MemFn memfn, oneshot_t one) {
        return [self = _NEFORCE move(self), memfn](Types... args) {
            auto ptr = signal::lock_if_weak(self);
            if (ptr == nullptr) {
                return callback_result::erase;
            }
            ((*ptr).*memfn)(_NEFORCE forward<Types>(args)...);
            return callback_result::erase;
        };
    }

    template <typename Self, typename MemFn>
    auto bind(Self self, MemFn memfn, nshot_t n) {
        return [self = _NEFORCE move(self), memfn, n = static_cast<size_t>(n)](Types... args) mutable {
            if (n == 0) {
                return callback_result::erase;
            }
            auto ptr = signal::lock_if_weak(self);
            if (ptr == nullptr) {
                return callback_result::erase;
            }
            ((*ptr).*memfn)(_NEFORCE forward<Types>(args)...);
            --n;
            if (n == 0) {
                return callback_result::erase;
            }
            return callback_result::keep;
        };
    }

    template <typename Func>
    enable_if_t<is_invocable_r_v<callback_result, Func, Types...>, connection> connect_impl(Func callback,
                                                                                            int priority) {
        connection conn{};
 
        auto it = slots_.begin();
        while (it != slots_.end() && it->priority >= priority) {
            ++it;
        }
        slots_.emplace(it, _NEFORCE move(callback), conn.flag(), priority);
 
        return conn;
    }

    template <typename Func>
    enable_if_t<!is_invocable_r_v<callback_result, Func, Types...>, connection> connect_impl(Func callback,
                                                                                             int priority) {
        connection conn{};
 
        auto wrapped = [callback = _NEFORCE move(callback)](Types... args) mutable {
            callback(_NEFORCE forward<Types>(args)...);
            return callback_result::keep;
        };
 
        auto it = slots_.begin();
        while (it != slots_.end() && it->priority >= priority) {
            ++it;
        }
        slots_.emplace(it, _NEFORCE move(wrapped), conn.flag(), priority);
 
        return conn;
    }
 
    size_t slot_count_unlocked() const noexcept {
        size_t count = 0;
        for (const auto& slot: slots_) {
            if (slot.connected_flag && *slot.connected_flag) {
                ++count;
            }
        }
        return count;
    }
 
    template <size_t... Is>
    void emit_dynamic_impl(const vector<reflect::meta_any>& args, index_sequence<Is...> /*unused*/) {
        this->emit(args[Is].template convert<Types>()...);
    }
 
public:
    signal() = default;
    ~signal() override = default;
    signal(const signal&) = delete;
    signal& operator=(const signal&) = delete;
    signal(signal&&) = delete;
    signal& operator=(signal&&) = delete;

    template <typename Self, typename MemFn, typename... Tag>
    connection connect(Self self, MemFn memfn, Tag... tag) {
        return this->connect(_NEFORCE move(self), memfn, 0, tag...);
    }

    template <typename Func>
    connection connect(Func callback, int priority = 0) {
        return this->with_lock([this, callback = _NEFORCE move(callback), priority] {
            return this->connect_impl(_NEFORCE move(callback), priority);
        });
    }

    template <typename Self, typename MemFn, typename... Tag>
    connection connect(Self self, MemFn memfn, int priority, Tag... tag) {
        static_assert(sizeof...(Tag) <= 1, "only zero or one tag is allowed");
 
        return this->with_lock([this, self = _NEFORCE move(self), memfn = _NEFORCE move(memfn), priority, tag...] {
            connection conn{};
 
            auto it = slots_.begin();
            while (it != slots_.end() && it->priority >= priority) {
                ++it;
            }
            slots_.emplace(it, this->bind(_NEFORCE move(self), memfn, tag...), conn.flag(), priority);
 
            return conn;
        });
    }

    void emit(Types... args) {
        vector<slot_entry> snapshot;
        {
            lock<mutex> lk(mutex_);
            if (is_blocked()) {
                return;
            }
            snapshot = slots_;
        }
 
        for (auto& slot: snapshot) {
            if (!slot.connected_flag || !(*slot.connected_flag)) {
                continue;
            }
 
            const callback_result res = slot.callback(args...);
            if (res == callback_result::erase) {
                *slot.connected_flag = false;
            }
        }
 
        {
            lock<mutex> lk(mutex_);
            slots_.erase(
                    _NEFORCE remove_if(slots_.begin(), slots_.end(),
                                       [](const slot_entry& s) { return !s.connected_flag || !(*s.connected_flag); }),
                    slots_.end());
        }
    }

    template <typename Executor>
    void emit_executor(Executor& executor, Types... args) {
        auto weak_flag = weak_ptr<bool>(blocked_flag_);
        auto args_tuple = _NEFORCE make_tuple(_NEFORCE forward<Types>(args)...);
 
        executor.post([this, weak_flag = _NEFORCE move(weak_flag), args_tuple = _NEFORCE move(args_tuple)]() mutable {
            if (weak_flag.expired()) {
                return;
            }
 
            _NEFORCE apply([this](auto&&... a) { this->emit(_NEFORCE forward<decltype(a)>(a)...); },
                           _NEFORCE move(args_tuple));
        });
    }

    template <typename... Args>
    void operator()(Args&&... args) {
        this->emit(_NEFORCE forward<Args>(args)...);
    }

    void disconnect_all() override {
        this->with_lock([this] { slots_.clear(); });
    }

    connection connect_signal(signal& other, int priority = 0) {
        return this->connect(
                [other_ptr = &other](Types... args) {
                    other_ptr->emit(args...);
                    return callback_result::keep;
                },
                priority);
    }

    connection connect_signal(signal* other, int priority = 0) {
        return this->connect(
                [other](Types... args) {
                    if (other) {
                        other->emit(args...);
                    }
                    return callback_result::keep;
                },
                priority);
    }

    template <typename Func, typename Predicate>
    connection connect_if(Func callback, Predicate pred, int priority = 0) {
        using result_type = invoke_result_t<Predicate, Types...>;
        static_assert(is_boolean_v<result_type>, "only boolean results are allowed");
 
        return this->connect(
                [callback = _NEFORCE move(callback), pred = _NEFORCE move(pred)](Types... args) mutable {
                    if (pred(args...)) {
                        callback(_NEFORCE forward<Types>(args)...);
                    }
                    return callback_result::keep;
                },
                priority);
    }

    template <typename Self, typename MemFn, typename Predicate, typename... Tag>
    connection connect_if(Self self, MemFn memfn, Predicate pred, int priority, Tag... tag) {
        using result_type = invoke_result_t<Predicate, Types...>;
        static_assert(is_boolean_v<result_type>, "only boolean results are allowed");
        static_assert(sizeof...(Tag) <= 1, "only zero or one tag is allowed");
 
        auto bound = this->bind(_NEFORCE move(self), memfn, tag...);
 
        return this->connect(
                [bound = _NEFORCE move(bound), pred = _NEFORCE move(pred)](Types... args) mutable {
                    if (pred(args...)) {
                        return bound(_NEFORCE forward<Types>(args)...);
                    }
                    return callback_result::keep;
                },
                priority);
    }

    template <typename Func, typename Filter>
    connection connect_filtered(Func callback, Filter filter, int priority = 0) {
        return this->connect(
                [callback = _NEFORCE move(callback), filter = _NEFORCE move(filter)](Types... args) mutable {
                    auto filtered = filter(args...);
                    if (filtered) {
                        auto filtered_tuple = _NEFORCE make_tuple(_NEFORCE move(*filtered));
                        _NEFORCE apply(
                                [&callback](auto&&... filtered_args) {
                                    callback(_NEFORCE forward<decltype(filtered_args)>(filtered_args)...);
                                },
                                _NEFORCE move(filtered_tuple));
                    }
                    return callback_result::keep;
                },
                priority);
    }

    template <typename Func, typename Transform>
    connection connect_transformed(Func callback, Transform transform, int priority = 0) {
        static_assert(is_invocable_v<Transform, Types...>, "only function inputs are allowed");
 
        return this->connect(
                [callback = _NEFORCE move(callback), transform = _NEFORCE move(transform)](Types... args) mutable {
                    callback(transform(_NEFORCE forward<Types>(args)...));
                    return callback_result::keep;
                },
                priority);
    }

    shared_ptr<bool> block_flag() const noexcept { return blocked_flag_; }

    bool is_blocked() const noexcept override { return blocked_flag_ && *blocked_flag_; }

    void block() override {
        if (blocked_flag_) {
            *blocked_flag_ = true;
        }
    }

    void unblock() override {
        if (blocked_flag_) {
            *blocked_flag_ = false;
        }
    }

    size_t slot_count() const noexcept override {
        return this->with_lock([this] { return slot_count_unlocked(); });
    }

    void connect_dynamic(function<void(const vector<reflect::meta_any>&)> callback) override {
        this->connect([cb = _NEFORCE move(callback)](Types... args) {
            vector<reflect::meta_any> wrapped;
            wrapped.reserve(sizeof...(Types));
            (wrapped.emplace_back(_NEFORCE forward<Types>(args)), ...);
            cb(wrapped);
            return callback_result::keep;
        });
    }

    void emit_dynamic(const vector<reflect::meta_any>& args) override {
        if (args.size() < sizeof...(Types)) {
            return;
        }
        emit_dynamic_impl(args, make_index_sequence<sizeof...(Types)>{});
    }

    NEFORCE_NODISCARD bool empty() const noexcept {
        return this->with_lock([this] { return slot_count_unlocked() == 0; });
    }
};
 // Signals
 
NEFORCE_END_NAMESPACE__
#endif // NEFORCE_CORE_ASYNC_SIGNALS_HPP__