render/template/include/parallel_event.hpp

/**
 * 线程安全版的event,可以用于多线程使用
 * 使用方法:
 *   1. 创建ParallelEvent对象
 *      ParallelEvent ev{};
 *   2. 建立连接,并保留句柄
 *      auto connection = ev.Subscribe( ... );
 *   3. 发射事件
 *      ev.EmitEvent( ... );
 *   4. 取消订阅, 直接析构connection句柄即可
 */

#ifndef ZCPPWIN_PARALLEL_EVENT_HPP
#define ZCPPWIN_PARALLEL_EVENT_HPP

#include <atomic>
#include <functional>
#include <mutex>
#include <unordered_map>
#include <vector>
#include <memory>
#include <type_traits>
#include <utility>

template<class... Args>
class ParallelEvent {
public:
    using Callback = std::function<void(Args...)>;

    class Connection { // 返回给客户的句柄,析构会Disconnect
    public:
        Connection() = default;
        Connection(const Connection &) = delete;
        Connection &operator=(const Connection &) = delete;

        Connection(Connection &&other) noexcept { MoveFrom(std::move(other)); }
        Connection &operator=(Connection &&other) noexcept {
            if (this != &other) {
                Disconnect();
                MoveFrom(std::move(other));
            }
            return *this;
        }

        ~Connection() {
            Disconnect();
        }

        void Disconnect() {
            if (owner_) {
                owner_->Unsubscribe(id_);
                owner_ = nullptr;
                id_ = 0;
            }
        }

        explicit operator bool() const noexcept { return owner_ != nullptr; }
    private:
        friend class ParallelEvent;
        Connection(ParallelEvent *owner, std::uint64_t id) : owner_(owner), id_(id) {}

        void MoveFrom(Connection &&other) noexcept {
            owner_ = other.owner_;
            id_ = other.id_;
            other.owner_ = nullptr;
            other.id_ = 0;
        }
    private:
        ParallelEvent *owner_ = nullptr;
        std::uint64_t id_ = 0;
    };


    ParallelEvent() = default;
    ParallelEvent(const ParallelEvent &) = delete;
    ParallelEvent &operator=(const ParallelEvent &) = delete;

    // 订阅
    // 例如：auto guard = shared_from_this(); event.subscribe(guard, [...]{});
    template<typename T>
    Connection Subscribe(std::weak_ptr<T> guard, Callback cb) {
        Handler h;
        h.id = nextId_++;
        h.guard = std::move(guard);
        h.cb = std::move(cb); //
        {
            std::lock_guard<std::mutex> lk(mtx_);
            handlers_.emplace(h.id, std::move(h));
        }
        return Connection(this, h.id);
    }
    Connection Subscribe(Callback cb) {
        Handler h;
        h.id = nextId_++;
        h.cb = std::move(cb); //
        {
            std::lock_guard lk(mtx_);
            handlers_.emplace(h.id, std::move(h));
        }
        return Connection(this, h.id);
    }

    // 订阅 + 指定派发器（想跨线程就传一个 dispatcher）
    // dispatcher 接收一个 “要执行的函数”，你可以用 invokeMethod / 线程池等方式调度
    template<typename T>
    Connection Subscribe(std::weak_ptr<T> guard, Callback cb,
                         std::function<void(std::function<void()>)> dispatcher) {
        Handler h;
        h.id = nextId_++;
        h.guard = std::move(guard);
        h.cb = std::move(cb);
        h.dispatcher = std::move(dispatcher); //
        {
            std::lock_guard lk(mtx_);
            handlers_.emplace(h.id, std::move(h));
        }
        return Connection(this, h.id);
    }

    // 发射事件（线程安全）
    void EmitEvent(Args... args) {
        // 1) 快照，避免回调里修改订阅造成死锁
        std::vector<Handler> snapshot;
        snapshot.reserve(64); //
        {
            std::lock_guard lk(mtx_);
            snapshot.reserve(handlers_.size());
            for (auto &kv: handlers_) snapshot.push_back(kv.second);
        }
        // 2) 调用（不持锁）
        bool needCleanup = false;
        for (Handler &h: snapshot) {
            if (!IsAlive(h)) {
                needCleanup = true;
                continue;
            }
            // 如果绑定了 dispatcher，优先走 dispatcher
            if (h.dispatcher) {
                std::function<void()> fn = MakeInvokeFn(h, args...);
                h.dispatcher(std::move(fn));
                continue;
            }
            // 非 QObject：直接调用
            h.cb(args...);
        }

        // 3) 懒清理：有失效订阅时，移除之（避免 map 越积越多）
        if (needCleanup) CleanupExpired();
    }

    // 便利别名
    void operator()(Args... args) { EmitEvent(std::forward<Args>(args)...); }

    // 主动清理全部失效订阅
    void CleanupExpired() {
        std::lock_guard lk(mtx_);
        for (auto it = handlers_.begin(); it != handlers_.end();) {
            if (!IsAlive(it->second)) it = handlers_.erase(it);
            else ++it;
        }
    }

    // 主动清空
    void Clear() {
        std::lock_guard lk(mtx_);
        handlers_.clear();
    }
private:
    struct Handler {
        std::uint64_t id = 0;
        std::weak_ptr<void> guard; // 生命周期守卫(可空)
        Callback cb;
        // 跨线程派发器（可选）
        std::function<void(std::function<void()>)> dispatcher;
    };

    void Unsubscribe(std::uint64_t id) {
        std::lock_guard lk(mtx_);
        handlers_.erase(id);
    }

    static bool IsAlive(const Handler &h) {
        if (!h.guard.expired()) {
            return true;
        }
        // 如果 guard 过期，认为失效
        // （允许“无 guard 订阅”，那就别用这个 overload）
        // 比较guard和nullptr的wp之间的控制块是否相同
        if (h.guard.owner_before(std::weak_ptr<void>{}) ||
            std::weak_ptr<void>{}.owner_before(h.guard)) {
            return !h.guard.expired();
        }
        // h.guard 为空 weak_ptr：当作永远存活（用户自行持有 Connection 管理）
        return true;
    }

    template<typename... A>
    static std::function<void()> MakeInvokeFn(const Handler &h, A &&... a) {
        // 捕获回调 + 参数副本，确保异步安全
        auto cb = h.cb;
        auto tup = std::make_tuple(std::forward<A>(a)...);
        return [cb = std::move(cb), tup = std::move(tup)]() mutable {
            std::apply(cb, tup);
        };
    }
private:
    std::mutex mtx_;
    std::unordered_map<std::uint64_t, Handler> handlers_;
    std::atomic<std::uint64_t> nextId_{1};
};



#endif //ZCPPWIN_PARALLEL_EVENT_HPP