broker_handler.h

#ifndef RECODEX_BROKER_BROKER_HANDLER_H
#define RECODEX_BROKER_BROKER_HANDLER_H

#include <spdlog/logger.h>

#include "../config/broker_config.h"
#include "../notifier/status_notifier.h"
#include "../reactor/command_holder.h"
#include "../reactor/handler_interface.h"
#include "../worker_registry.h"

class broker_handler : public handler_interface
{
public:
    broker_handler(std::shared_ptr<const broker_config> config,
        std::shared_ptr<worker_registry> workers,
        std::shared_ptr<spdlog::logger> logger);

    void on_request(const message_container &message, response_cb respond);

private:
    std::shared_ptr<const broker_config> config_;

    std::shared_ptr<worker_registry> workers_;

    std::shared_ptr<spdlog::logger> logger_;

    std::map<worker_registry::worker_ptr, std::chrono::milliseconds> worker_timers_;

    command_holder worker_commands_;

    command_holder client_commands_;

    void process_worker_init(const std::string &identity, const std::vector<std::string> &message, response_cb respond);

    void process_worker_done(const std::string &identity, const std::vector<std::string> &message, response_cb respond);

    void process_worker_ping(const std::string &identity, const std::vector<std::string> &message, response_cb respond);

    void process_worker_progress(
        const std::string &identity, const std::vector<std::string> &message, response_cb respond);

    void process_client_eval(const std::string &identity, const std::vector<std::string> &message, response_cb respond);

    void process_timer(const message_container &message, response_cb respond);

    bool reassign_request(worker::request_ptr request, response_cb respond);

    bool assign_queued_request(worker_registry::worker_ptr worker, response_cb respond);

    bool check_failure_count(worker::request_ptr request, status_notifier_interface &status_notifier);
};

#endif // RECODEX_BROKER_BROKER_HANDLER_H