service_discovery_processor.h

Go to the documentation of this file.

// Copyright (c) 2018, Smart Projects Holdings Ltd
// All rights reserved.
// See LICENSE file for license details.
#ifndef _UGCS_VSM_SERVICE_DISCOVERY_PROCESSOR_H_
#define _UGCS_VSM_SERVICE_DISCOVERY_PROCESSOR_H_

#include <ugcs/vsm/request_worker.h>
#include <ugcs/vsm/socket_processor.h>
#include <ugcs/vsm/callback.h>
#include <set>

namespace ugcs {
namespace vsm {

class Service_discovery_processor : public Request_processor
{
    DEFINE_COMMON_CLASS(Service_discovery_processor, Request_container);
public:
    Service_discovery_processor(
        Socket_address::Ptr muticast_adress =
            Socket_address::Create(DEFAULT_DISCOVERY_ADDRESS, DEFAULT_DISCOVERY_PORT));

    virtual ~Service_discovery_processor();

    typedef Callback_proxy<void, std::string, std::string, std::string, std::string, bool> Detection_handler;

    DEFINE_CALLBACK_BUILDER(
            Make_detection_handler,
            (std::string, std::string, std::string, std::string, bool),
            ("", "", "", "", true));

    void
    Advertise_service(const std::string& type, const std::string& name, const std::string& location);

    void
    Unadvertise_service(const std::string& type, const std::string& name, const std::string& location);

    void
    Subscribe_for_service(
            const std::string &type,
            Detection_handler handler,
            Request_processor::Ptr context);

    void
    Search_for_service(const std::string& type);

    void
    Unsubscribe_from_service(const std::string& type);

    template <typename... Args>
    static Ptr
    Get_instance(Args &&... args)
    {
        return singleton.Get_instance(std::forward<Args>(args)...);
    }

private:
    static std::string DEFAULT_DISCOVERY_ADDRESS;
    static std::string DEFAULT_DISCOVERY_PORT;
    static std::string LOOPBACK_BROADCAST_ADDRESS;
    static std::string SEARCH_METHOD_STRING;
    static std::string NOTIFY_METHOD_STRING;
    // token to replace with interface IP in notify message and discovery response
    static std::string LOCAL_ADDRESS_IDENTIFIER;
    // used to distinguish loopback socket in reader callback
    static std::string LOOPBACK_IDENTIFIER;
    // used to distinguish multicast socket in reader callback
    static std::string MC_IDENTIFIER;

    // Randomly generated string to uniquely identify service instance.
    // Used by service locator to distinguish between multiple locations
    // for the same service instance.
    std::string my_instance_identifier;

    static Singleton<Service_discovery_processor> singleton;

    // Service type, name, location
    typedef std::tuple<std::string, std::string, std::string> Service_info;

    // Sender socket with current io operation.
    typedef std::pair<Operation_waiter, Socket_processor::Stream::Ref> My_socket;

    // Multicasts listener address
    Socket_address::Ptr multicast_adress;

    // loopback detection requests are sent to this address so they are
    // received by all locally listening processes.
    Socket_address::Ptr loopback_broadcast_adress;

    // Sender and reader callbacks executing in this thread.
    Request_worker::Ptr worker;

    // Listener socket bound on 0.0.0.0 on multicast_port.
    My_socket receiver;

    // One socket per interface indexed by ip address string.
    // Bound to that interface IP.
    // Used to send discovery packets.
    // Incoming unicast responses are read from this socket.
    std::map<std::string, My_socket> sender_sockets;

    // Loopback interface handled separately. Bound to 127.0.0.1:0
    My_socket sender_loopback;

    // advertised services
    std::set<Service_info> my_services;

    // Services to listen for.
    // maps service type into user specified callback
    std::map<std::string, std::pair<Detection_handler, Request_processor::Ptr>> subscribed_services;

    // look for new interfaces each 5 seconds.
    Timer_processor::Timer::Ptr interface_checker_timer;

    // Send notifications each 10 seconds.
    Timer_processor::Timer::Ptr notify_timer;

    void
    On_read(
            Io_buffer::Ptr buffer,
            Io_result result,
            Socket_address::Ptr addr,
            std::string stream_id);

    bool
    On_timer();

    bool
    On_notify_timer();

    void
    On_sender_bound(Socket_processor::Stream::Ref l, Io_result);

    void
    Send_notify(
            Socket_processor::Stream::Ref s,
            Socket_address::Ptr dest_addr,
            const std::string& type,
            const std::string& name,
            const std::string& location,
            bool alive);

    // Sends notifications about all services on given interface
    void
    Send_notify_all_services(
        Socket_processor::Stream::Ref s,
        Socket_address::Ptr dest_addr);

    void
    Send_response(
            Socket_address::Ptr addr,
            const std::string& type,
            const std::string& name,
            const std::string& location);

    void
    Send_msearch(
            Socket_processor::Stream::Ref s,
            Socket_address::Ptr dest_addr,
            const std::string& type);

    virtual void
    On_enable() override;

    // return true  if already activated
    //        false if was not activated
    bool
    Activate();

    // close all sockets when there are no subcriptions or
    void
    Deactivate_if_no_services();

    void
    On_advertise(
            const std::string type,
            const std::string name,
            const std::string location,
            Request::Ptr request);

    void
    On_unadvertise(
            const std::string type,
            const std::string name,
            const std::string location,
            Request::Ptr request);

    void
    On_subscribe_for_service(
            const std::string type,
            Detection_handler handler,
            Request_processor::Ptr context,
            Request::Ptr request);

    void
    On_unsubscribe_from_service(
            const std::string type,
            Request::Ptr request);

    void
    On_search_for_service(
            const std::string type,
            Request::Ptr request);

    void
    On_disable_impl(Request::Ptr request);

    virtual void
    On_disable() override;

    void
    Schedule_read(std::string id);

    static bool
    Has_location_string(const std::string& loc);

    static std::string
    Build_location_string(const std::string& loc, Socket_address::Ptr local_addr);
};

} /* namespace vsm */
} /* namespace ugcs */

#endif /* _UGCS_VSM_SERVICE_DISCOVERY_PROCESSOR_H_ */