packet-dispatcher.hpp

Go to the documentation of this file.

#pragma once
#include "packet.hpp"
 
namespace mal_packet_weaver
{
    template <typename DerivedPacket, typename... Args>
    using PacketHandlerFunc = std::function<void(Args..., std::unique_ptr<DerivedPacket>)>;
 
    template <typename DerivedPacket, typename... Args>
    using PacketFilterFunc = std::function<bool(Args..., DerivedPacket const &)>;
 
    class PacketDispatcher final : public non_copyable_non_movable,
                                   public std::enable_shared_from_this<PacketDispatcher>
    {
        template <typename Value>
        class SynchronizationWrapper
        {
        public:
            SynchronizationWrapper(boost::asio::io_context &context, std::function<void(std::vector<Value> &)> fn)
                : strand_{ context }, dequeue_function_(std::move(fn))
            {
            }
            inline void push(Value &&value) requires(std::is_copy_constructible_v<Value> || std::is_copy_assignable_v<Value>)
            {
                strand_.post(
                    [this, copied_value = value]()
                    {
                        input_data_.emplace_back(std::move(copied_value));
                        synchronization_flag_.test_and_set(std::memory_order_release);
                    }
                );
            }
            inline void push(Value &&value)
                requires(std::is_same_v<Value, std::unique_ptr<typename Value::element_type>>)
            {
                strand_.post(
                    [this, released_ptr = value.release()]()
                    {
                        Value unique_ptr{ released_ptr };
                        input_data_.emplace_back(std::move(unique_ptr));
                        synchronization_flag_.test_and_set(std::memory_order_release);
                    });
            }
 
            inline bool has_data() { return synchronization_flag_.test(std::memory_order_acquire); }
 
            inline std::future<bool> create_pop_task()
            {
                std::shared_ptr<std::promise<bool>> promise = std::make_shared<std::promise<bool>>();
                std::future<bool> input_future = promise->get_future();
                strand_.post(
                    [this, promise]()
                    {
                        if (input_data_.empty())
                        {
                            promise->set_value(false);
                            synchronization_flag_.clear(std::memory_order_release);
                            return;
                        }
                        dequeue_function_(input_data_);
                        input_data_.clear();
                        promise->set_value(true);
                        synchronization_flag_.clear(std::memory_order_release);
                    });
                return input_future;
            }
 
        private:
            boost::asio::io_context::strand strand_; 
            std::atomic_flag synchronization_flag_;  
            std::function<void(std::vector<Value> &)> dequeue_function_; 
            std::vector<Value> input_data_;                              
        };
 
    public:
        using BasePacketPtr = std::unique_ptr<Packet>;
 
        template <typename T>
        using shared_promise = std::shared_ptr<std::promise<T>>;
 
        using shared_packet_promise = shared_promise<BasePacketPtr>;
 
        using promise_filter = std::pair<std::function<bool(BasePacketPtr const &)>, shared_packet_promise>;
 
        using handler_tuple = std::tuple<float, PacketFilterFunc<Packet>, PacketHandlerFunc<Packet>>;
 
        PacketDispatcher(boost::asio::io_context &io_context);
 
        inline void enqueue_packet(BasePacketPtr &&packet);
        template <IsPacket DerivedPacket>
        boost::asio::awaitable<std::unique_ptr<DerivedPacket>> await_packet(float timeout = -1.0f);
        template <IsPacket DerivedPacket>
        boost::asio::awaitable<std::unique_ptr<DerivedPacket>> await_packet(PacketFilterFunc<DerivedPacket> filter,
                                                                            float timeout = -1.0f);
        template <IsPacket DerivedPacket>
        void register_default_handler(PacketHandlerFunc<DerivedPacket> handler,
                                      PacketFilterFunc<DerivedPacket> filter = {}, float delay = 0.0f);
 
        void register_subsystem_handler(PacketSubsystemID subsystem_id, PacketHandlerFunc<Packet> handler,
                                        PacketFilterFunc<Packet> filter = {}, float delay = 0.0f);
 
        inline void enqueue_promise(UniquePacketID packet_id, shared_packet_promise promise);
        inline void enqueue_filter_promise(UniquePacketID packet_id, promise_filter filtered_promise);
 
        void Destroy() { alive_.store(false); }
 
    private:
        boost::asio::awaitable<std::shared_ptr<PacketDispatcher>> get_shared_ptr();
 
        boost::asio::awaitable<void> Run();
 
        inline bool fulfill_promises(UniquePacketID packet_id, BasePacketPtr &packet);
 
        inline bool fulfill_handlers(UniquePacketID packet_id, BasePacketPtr &packet, float &min_handler_timestamp,
                                     SteadyTimer &timer);
 
        inline void push_packet(BasePacketPtr &&packet);
 
        boost::asio::awaitable<bool> pop_inputs();
 
        boost::asio::io_context &io_context_; 
        SignalHandler signal_handler_;        
        SynchronizationWrapper<BasePacketPtr>
            unprocessed_packets_input_; 
        SynchronizationWrapper<std::pair<UniquePacketID, shared_packet_promise>>
            promise_map_input_; 
        SynchronizationWrapper<std::pair<UniquePacketID, promise_filter>>
            promise_filter_map_input_; 
        SynchronizationWrapper<std::pair<UniquePacketID, handler_tuple>>
            default_handlers_input_; 
        SynchronizationWrapper<std::pair<PacketSubsystemID, handler_tuple>>
            subsystem_handlers_input_; 
        std::unordered_map<UniquePacketID, std::vector<BasePacketPtr>>
            unprocessed_packets_; 
        std::unordered_map<UniquePacketID, std::deque<shared_packet_promise>>
            promise_map_; 
        std::unordered_map<UniquePacketID, std::vector<promise_filter>>
            promise_filter_map_; 
        std::unordered_map<UniquePacketID, std::vector<handler_tuple>>
            default_handlers_; 
        std::unordered_map<PacketSubsystemID, std::vector<handler_tuple>>
            subsystem_handlers_; 
        std::atomic_bool alive_{ true };
    };
}  // namespace mal_packet_weaver
 
#include "packet-dispatcher.inl"