resolver.h

#ifndef B0__RESOLVER__RESOLVER_H__INCLUDED
#define B0__RESOLVER__RESOLVER_H__INCLUDED

#include <b0/node.h>
#include <b0/protobuf/service_server.h>
#include <b0/protobuf/publisher.h>

#include <string>
#include <vector>
#include <set>
#include <boost/thread.hpp>
#include <boost/format.hpp>

namespace b0
{

namespace resolver
{


struct ServiceEntry;

struct NodeEntry
{
    std::string host_id;
    int process_id;
    std::string thread_id;
    std::string name;
    boost::posix_time::ptime last_heartbeat;
    std::vector<ServiceEntry*> services;
};

struct ServiceEntry
{
    NodeEntry *node;
    std::string name;
    std::string addr;
};

class Resolver;

class ResolverServiceServer : public b0::protobuf::ServiceServer<b0::resolver_msgs::Request, b0::resolver_msgs::Response>
{
public:
    ResolverServiceServer(Resolver *resolver);

    void setPort(int port);

    int port() const;

protected:
    virtual void announce() override;

    Resolver *resolver_;

    int port_;
};


class Resolver : public Node
{
public:
    Resolver();

    virtual ~Resolver();

    void init() override;

    void shutdown() override;

    virtual std::string getXPUBSocketAddress() const override;

    virtual std::string getXSUBSocketAddress() const override;

    virtual void announceNode() override;

    virtual void notifyShutdown() override;

    void onNodeConnected(std::string name);

    void onNodeDisconnected(std::string name);

    void onNodeTopicPublishStart(std::string node_name, std::string topic_name);

    void onNodeTopicPublishStop(std::string node_name, std::string topic_name);

    void onNodeTopicSubscribeStart(std::string node_name, std::string topic_name);

    void onNodeTopicSubscribeStop(std::string node_name, std::string topic_name);

    void onNodeServiceOfferStart(std::string node_name, std::string service_name);

    void onNodeServiceOfferStop(std::string node_name, std::string service_name);

    void onNodeServiceUseStart(std::string node_name, std::string service_name);

    void onNodeServiceUseStop(std::string node_name, std::string service_name);

    void pubProxy(int xsub_proxy_port, int xpub_proxy_port);

    bool nodeNameExists(std::string name);

    virtual void setResolverPort(int port);

    virtual std::string address(std::string host, int port);

    virtual std::string address(int port);

    virtual resolver::NodeEntry * nodeByName(std::string node_name);

    virtual resolver::ServiceEntry * serviceByName(std::string service_name);

    virtual void heartBeat(resolver::NodeEntry *node_entry);

    virtual void handle(const b0::resolver_msgs::Request &req, b0::resolver_msgs::Response &resp);

    std::string makeUniqueNodeName(std::string nodeName);

    virtual void handleAnnounceNode(const b0::resolver_msgs::AnnounceNodeRequest &rq, b0::resolver_msgs::AnnounceNodeResponse &rsp);

    virtual void handleShutdownNode(const b0::resolver_msgs::ShutdownNodeRequest &rq, b0::resolver_msgs::ShutdownNodeResponse &rsp);

    virtual void handleAnnounceService(const b0::resolver_msgs::AnnounceServiceRequest &rq, b0::resolver_msgs::AnnounceServiceResponse &rsp);

    virtual void handleResolveService(const b0::resolver_msgs::ResolveServiceRequest &rq, b0::resolver_msgs::ResolveServiceResponse &rsp);

    virtual void handleHeartBeat(const b0::resolver_msgs::HeartBeatRequest &rq, b0::resolver_msgs::HeartBeatResponse &rsp);

    void handleNodeTopic(const b0::resolver_msgs::NodeTopicRequest &req, b0::resolver_msgs::NodeTopicResponse &resp);

    void handleNodeService(const b0::resolver_msgs::NodeServiceRequest &req, b0::resolver_msgs::NodeServiceResponse &resp);

    void handleGetGraph(const b0::resolver_msgs::GetGraphRequest &req, b0::resolver_msgs::GetGraphResponse &resp);

    void getGraph(b0::resolver_msgs::Graph &graph);

    void onGraphChanged();

    void heartBeatSweeper();

    void setOnlineMonitoring(bool enabled);

protected:
    ResolverServiceServer resolv_server_;

    std::string xsub_proxy_addr_;

    std::string xpub_proxy_addr_;

    boost::thread pub_proxy_thread_;

    boost::thread heartbeat_sweeper_thread_;

    std::map<std::string, resolver::NodeEntry*> nodes_by_name_;

    std::map<std::string, resolver::NodeEntry*> nodes_by_key_;

    std::map<std::string, resolver::ServiceEntry*> services_by_name_;

    std::set<std::pair<std::string, std::string> > node_publishes_topic_;

    std::set<std::pair<std::string, std::string> > node_subscribes_topic_;

    std::set<std::pair<std::string, std::string> > node_offers_service_;

    std::set<std::pair<std::string, std::string> > node_uses_service_;

    b0::protobuf::Publisher<b0::resolver_msgs::Graph> graph_pub_;

    bool online_monitoring_;
};

} // namespace resolver

} // namespace b0

#endif // B0__RESOLVER__RESOLVER_H__INCLUDED