EventBus.hpp

Go to the documentation of this file.

 
#pragma once
 
#include "corium/EventSink.hpp"
#include "corium/internal/CallableTraits.hpp"
#include "corium/internal/EventQueue.hpp"
#include "corium/internal/Reactor.hpp"
#include "corium/profiler/ProfilerPolicies.hpp"
 
#include <utility>
 
namespace corium {
 
template <
    typename EventVariantType = DefaultEvents,
    typename QueuePolicy = BoundedMpscQueuePolicy<EventVariantType, 1024>,
    typename SignalPolicy = NoSignalPolicy,
    typename StoragePolicy = DefaultStoragePolicy,
    typename OverflowPolicy = DropNewestOverflowPolicy,
    typename ProfilerPolicy = profiler::NullProfiler
>
class BasicEventBus {
public:
    using EventVariant = EventVariantType;
    using ReactorType = BasicReactor<EventVariant, StoragePolicy>;
    using ProfilerPolicyType = ProfilerPolicy;
 
    BasicEventBus() = default;
 
    void post(EventVariant&& event, EventPriority priority = EventPriority::Normal)
    {
        _profilerPolicy.onEventPosted(event, static_cast<uint8_t>(priority));
        _profilerPolicy.recordPostTime(_profilerPolicy.nowNs());
        _eventQueue.pushEvent(std::move(event), priority);
    }
 
    void post(const EventVariant& event, EventPriority priority = EventPriority::Normal)
    {
        _profilerPolicy.onEventPosted(event, static_cast<uint8_t>(priority));
        _profilerPolicy.recordPostTime(_profilerPolicy.nowNs());
        _eventQueue.pushEvent(event, priority);
    }
 
    template <typename Event>
    void postHighPriority(Event&& event)
    {
        post(EventVariant(std::forward<Event>(event)), EventPriority::High);
    }
 
    bool processOne()
    {
        EventVariant event;
        if (!_eventQueue.tryPopEvent(event)) {
            return false;
        }
        const uint64_t postTime     = _profilerPolicy.takePostTime();
        const uint64_t dispatchTime = _profilerPolicy.nowNs();
        _reactor.dispatch(event);
        const uint64_t finishTime = _profilerPolicy.nowNs();
        _profilerPolicy.onEventDispatched(event, 0, postTime, dispatchTime, finishTime);
        return true;
    }
 
    std::size_t processBatch(std::size_t maxBatch)
    {
        std::size_t count = 0;
        EventVariant event;
        while (count < maxBatch) {
            if (!_eventQueue.tryPopEvent(event)) {
                break;
            }
            const uint64_t postTime     = _profilerPolicy.takePostTime();
            const uint64_t dispatchTime = _profilerPolicy.nowNs();
            _reactor.dispatch(event);
            const uint64_t finishTime = _profilerPolicy.nowNs();
            _profilerPolicy.onEventDispatched(event, 0, postTime, dispatchTime, finishTime);
            count++;
        }
        return count;
    }
 
    std::size_t drain()
    {
        std::size_t total = 0;
        while (processOne()) {
            total++;
        }
        return total;
    }
 
    [[nodiscard]] ProfilerPolicy& profiler() noexcept
    {
        return _profilerPolicy;
    }
 
    [[nodiscard]] const ProfilerPolicy& profiler() const noexcept
    {
        return _profilerPolicy;
    }
 
    [[nodiscard]] bool empty() const
    {
        return _eventQueue.empty();
    }
 
    void seal()
    {
        _reactor.seal();
    }
 
    void setOnQueueNonEmpty(StaticCallback callback)
    {
        _eventQueue.setOnQueueNonEmpty(callback);
    }
 
    template <typename EventType, typename Handler>
    bool registerHandler(Handler&& handler)
    {
        return _reactor.template registerHandler<EventType>(std::forward<Handler>(handler));
    }
 
    template <typename Handler>
    bool registerHandler(Handler&& handler)
    {
        using EventType = callable_event_type_t<Handler>;
        return _reactor.template registerHandler<EventType>(std::forward<Handler>(handler));
    }
 
    [[nodiscard]] SignalPolicy& signalPolicy() noexcept
    {
        return _eventQueue.signalPolicy();
    }
 
    [[nodiscard]] const SignalPolicy& signalPolicy() const noexcept
    {
        return _eventQueue.signalPolicy();
    }
 
    [[nodiscard]] OverflowPolicy& overflowPolicy() noexcept
    {
        return _eventQueue.overflowPolicy();
    }
 
    [[nodiscard]] const OverflowPolicy& overflowPolicy() const noexcept
    {
        return _eventQueue.overflowPolicy();
    }
 
    ReactorType& reactor() noexcept
    {
        return _reactor;
    }
 
    EventSinkT<EventVariant> sink() noexcept
    {
        return EventSinkT<EventVariant>(*this);
    }
 
private:
    EventQueue<QueuePolicy, SignalPolicy, OverflowPolicy> _eventQueue;
    ReactorType _reactor;
    ProfilerPolicy _profilerPolicy{};
};
 
using EventBus = BasicEventBus<DefaultEvents>;
 
template <
    typename EventVariantType = DefaultEvents,
    typename QueuePolicy = BoundedMpscQueuePolicy<EventVariantType, 1024>,
    typename SignalPolicy = NoSignalPolicy,
    typename StoragePolicy = DefaultStoragePolicy,
    typename OverflowPolicy = DropNewestOverflowPolicy,
    typename ProfilerPolicy = profiler::NullProfiler
>
using EventBusT = BasicEventBus<EventVariantType, QueuePolicy, SignalPolicy, StoragePolicy, OverflowPolicy, ProfilerPolicy>;
 
} // namespace corium