reactor.h

#ifndef RECODEX_BROKER_REACTOR_H
#define RECODEX_BROKER_REACTOR_H

#include <atomic>
#include <map>
#include <memory>
#include <zmq.hpp>

#include "handler_interface.h"
#include "message_container.h"
#include "socket_wrapper_base.h"

/* Forward */
class reactor;

class handler_wrapper
{
public:
    handler_wrapper(reactor &reactor_ref, std::shared_ptr<handler_interface> handler);

    virtual ~handler_wrapper();

    virtual void operator()(const message_container &message);

protected:
    std::shared_ptr<handler_interface> handler_;

    reactor &reactor_;
};

class asynchronous_handler_wrapper : public handler_wrapper
{
public:
    asynchronous_handler_wrapper(zmq::context_t &context,
        zmq::socket_t &async_handler_socket,
        reactor &reactor_ref,
        std::shared_ptr<handler_interface> handler);

    ~asynchronous_handler_wrapper();

    virtual void operator()(const message_container &message);

private:
    static const std::string TERMINATE_MSG;

    zmq::context_t &context_;

    zmq::socket_t &reactor_socket_;

    std::thread worker_;

    void handler_thread();

    void send_response(zmq::socket_t &socket, const message_container &message);

    const std::string unique_id_;
};

class reactor
{
public:
    const static std::string KEY_TIMER;

    const std::string unique_id;

    reactor(std::shared_ptr<zmq::context_t> context);

    void add_socket(const std::string &name, std::shared_ptr<socket_wrapper_base> socket);

    void add_handler(const std::vector<std::string> &origins, std::shared_ptr<handler_interface> handler);

    void add_async_handler(const std::vector<std::string> &origins, std::shared_ptr<handler_interface> handler);

    void send_message(const message_container &message);

    void process_message(const message_container &message);

    void start_loop();

    void terminate();

private:
    std::map<std::string, std::shared_ptr<socket_wrapper_base>> sockets_;

    std::multimap<std::string, std::shared_ptr<handler_wrapper>> handlers_;

    std::shared_ptr<zmq::context_t> context_;

    zmq::socket_t async_handler_socket_;

    std::atomic<bool> termination_flag_;
};

#endif // RECODEX_BROKER_REACTOR_H