dispatcher.h

Go to the documentation of this file.

#pragma once
 
#include "bcnp/stream_parser.h"
#include "bcnp/message_queue.h"
#include <bcnp/message_types.h>
#include <chrono>
#include <cstdint>
#include <functional>
#include <mutex>
#include <unordered_map>
 
namespace bcnp {
 
struct DispatcherConfig {
    std::size_t parserBufferSize{4096};
    std::chrono::milliseconds connectionTimeout{200};
};
 
using PacketHandler = std::function<void(const PacketView&)>;
 
using ErrorHandler = std::function<void(const StreamParser::ErrorInfo&)>;
 
class PacketDispatcher {
public:
    using Clock = std::chrono::steady_clock;
 
    explicit PacketDispatcher(DispatcherConfig config = {});
 
    void PushBytes(const uint8_t* data, std::size_t length);
 
    template<typename MsgType>
    void RegisterHandler(PacketHandler handler) {
        RegisterHandler(MsgType::kTypeId, std::move(handler));
    }
 
    void RegisterHandler(MessageTypeId typeId, PacketHandler handler);
 
    void UnregisterHandler(MessageTypeId typeId);
 
    void SetErrorHandler(ErrorHandler handler);
 
    bool IsConnected(Clock::time_point now) const;
 
    Clock::time_point LastReceiveTime() const;
 
    StreamParser& Parser() { return m_parser; }
    const StreamParser& Parser() const { return m_parser; }
 
    void SetWireSizeLookup(StreamParser::WireSizeLookup lookup) {
        m_parser.SetWireSizeLookup(std::move(lookup));
    }
    
    template<typename... MsgTypes>
    void RegisterMessageTypes() {
        m_parser.SetWireSizeLookup([](MessageTypeId typeId) -> std::size_t {
            return GetWireSizeFor<MsgTypes...>(typeId);
        });
    }
 
    uint64_t ParseErrorCount() const;
 
private:
    template<typename First, typename... Rest>
    static std::size_t GetWireSizeFor(MessageTypeId typeId) {
        if (typeId == First::kTypeId) {
            return First::kWireSize;
        }
        if constexpr (sizeof...(Rest) > 0) {
            return GetWireSizeFor<Rest...>(typeId);
        }
        return 0;
    }
 
    void HandlePacket(const PacketView& packet);
    void HandleError(const StreamParser::ErrorInfo& error);
 
    DispatcherConfig m_config;
    StreamParser m_parser;
    mutable std::mutex m_mutex;
    std::unordered_map<uint16_t, PacketHandler> m_handlers;
    ErrorHandler m_errorHandler;
    Clock::time_point m_lastRx{Clock::time_point::min()};
    uint64_t m_parseErrors{0};
};
 
// Type alias for backward compatibility
using MessageHandler = PacketHandler;
 
} // namespace bcnp