listener.h

Go to the documentation of this file.

 
#ifndef QB_IO_ASYNC_LISTENER_H_
#define QB_IO_ASYNC_LISTENER_H_
 
#include <algorithm>
#include <qb/system/container/unordered_set.h>
#include <qb/utility/branch_hints.h>
#include <qb/utility/type_traits.h>
#include <thread>
#include <vector>
#include "event/base.h"
 
namespace qb::io::async {

class listener {
public:
    thread_local static listener current;

    template <typename _Event, typename _Actor>
    class RegisteredKernelEvent final : public IRegisteredKernelEvent {
        friend class listener;
 
        _Actor &_actor; 
        _Event  _event; 

        ~RegisteredKernelEvent() final {
            _event.stop();
        }

        explicit RegisteredKernelEvent(ev::loop_ref loop, _Actor &actor) noexcept
            : _actor(actor)
            , _event(loop) {}

        void
        invoke() final {
            if constexpr (has_member_func_is_alive<_Actor>::value) {
                if (likely(_actor.is_alive()))
                    _actor.on(_event);
            } else
                _actor.on(_event);
        }
    };
 
private:
    ev::dynamic_loop _loop; 
    qb::unordered_set<IRegisteredKernelEvent *>
                _registeredEvents;      
    std::size_t _nb_invoked_events = 0; 
 
public:
    listener()
        : _loop(EVFLAG_AUTO) {}

    void
    clear() {
        if (!_registeredEvents.empty()) {
            for (auto it : _registeredEvents)
                delete it;
            _registeredEvents.clear();
            run(EVRUN_ONCE);
        }
    }

    ~listener() noexcept {
        clear();
    }

    template <typename EV_EVENT>
    void
    on(EV_EVENT &event, int revents) {
        auto &w    = *reinterpret_cast<event::base<EV_EVENT> *>(&event);
        w._revents = revents;
        w._interface->invoke();
        ++_nb_invoked_events;
    }

    template <typename _Event, typename _Actor, typename... _Args>
    _Event &
    registerEvent(_Actor &actor, _Args &&...args) {
        auto revent = new RegisteredKernelEvent<_Event, _Actor>(_loop, actor);
        revent->_event.template set<listener, &listener::on<typename _Event::ev_t>>(
            this);
        revent->_event._interface = revent;
 
        if constexpr (sizeof...(_Args) > 0)
            revent->_event.set(std::forward<_Args>(args)...);
 
        _registeredEvents.emplace(revent);
        return revent->_event;
    }

    void
    unregisterEvent(IRegisteredKernelEvent *kevent) {
        _registeredEvents.erase(kevent);
        delete kevent;
    }

    [[nodiscard]] inline ev::loop_ref
    loop() const {
        return _loop;
    }

    inline void
    run(int flag = 0) {
        _nb_invoked_events = 0;
        _loop.run(flag);
    }

    inline void
    break_one() {
        _loop.break_loop();
    }

    [[nodiscard]] inline std::size_t
    nb_invoked_event() const {
        return _nb_invoked_events;
    }

    [[nodiscard]] inline std::size_t
    size() const {
        return _registeredEvents.size();
    }
};

inline void
init() {
    // listener::current.clear();
}

inline std::size_t
run(int flag = 0) {
    listener::current.run(flag);
    return listener::current.nb_invoked_event();
}

inline std::size_t
run_once() {
    listener::current.run(EVRUN_ONCE);
    return listener::current.nb_invoked_event();
}

inline std::size_t
run_until(bool const &status) {
    std::size_t nb_invoked_events = 0;
    while (status) {
        listener::current.run(EVRUN_NOWAIT);
        nb_invoked_events += listener::current.nb_invoked_event();
    }
    return nb_invoked_events;
}

inline void
break_parent() {
    listener::current.break_one();
}
 
} // namespace qb::io::async
 
#endif // QB_IO_ASYNC_LISTENER_H_