SocketClient.hxx

Go to the documentation of this file.

 
#ifndef _UTILS_SOCKET_CLIENT_HXX_
#define _UTILS_SOCKET_CLIENT_HXX_
 
#include <functional>
#include <netdb.h>
#ifndef ESP_PLATFORM
// this doesn't exist on the ESP32 with LWiP
#include <arpa/inet.h>
#endif
#include <fcntl.h>
#include <ifaddrs.h>
#include <array>
 
#include "executor/StateFlow.hxx"
#include "executor/Timer.hxx"
#include "os/MDNS.hxx"
#include "os/sleep.h"
#include "utils/Atomic.hxx"
#include "utils/SocketClientParams.hxx"
#include "utils/format_utils.hxx"
 
class SocketClient : public StateFlowBase, private Atomic
{
public:
    SocketClient(Service *service, ExecutorBase *connect_executor,
        ExecutorBase *mdns_executor, std::unique_ptr<SocketClientParams> params,
        std::function<void(int, Notifiable *)> connect_callback)
        : StateFlowBase(service)
        , callback_(connect_callback)
        , connectExecutor_(connect_executor)
        , mdnsExecutor_(mdns_executor)
        , strategyOffset_(0)
        , mdnsPending_(false)
        , mdnsJoin_(false)
        , sleeping_(false)
        , requestShutdown_(false)
        , isConnected_(false)
        , fd_(-1)
    {
        reset_params(std::move(params));
        start_flow(STATE(start_connection));
    }
 
    ~SocketClient()
    {
        shutdown();
    }
 
    struct AddrInfoDeleter
    {
        void operator()(struct addrinfo *s)
        {
            if (s)
            {
                freeaddrinfo(s);
            }
        }
    };
 
    typedef std::unique_ptr<struct addrinfo, AddrInfoDeleter> AddrinfoPtr;
 
    enum class Attempt : uint8_t
    {
        RECONNECT,
        INITIATE_MDNS,
        CONNECT_MDNS,
        CONNECT_STATIC,
        WAIT_RETRY,
        FAILED_EXIT,
    };
 
    void reset_params(std::unique_ptr<SocketClientParams> params)
    {
        params_ = std::move(params);
    }
 
    void shutdown()
    {
        LOG(VERBOSE,
            "socketclient: sync shutdown isconn %d slp %d rqshut %d mdnsP %d"
            "mdnsJ %d isShut %d",
            isConnected_, sleeping_, requestShutdown_, mdnsPending_, mdnsJoin_,
            is_shutdown());
        start_shutdown();
        while (true)
        {
            AtomicHolder h(this);
            if (!mdnsPending_)
            {
                break;
            }
        }
        while (!is_terminated())
        {
            microsleep(1000);
        }
    }
 
    bool is_connected()
    {
        AtomicHolder h(this);
        return isConnected_;
    }
 
    bool is_shutdown()
    {
        return is_terminated();
    }
 
    void start_shutdown()
    {
        LOG(VERBOSE,
            "socketclient: start shutdown isconn %d slp %d rqshut %d mdnsP %d"
            "mdnsJ %d isShut %d",
            isConnected_, sleeping_, requestShutdown_, mdnsPending_, mdnsJoin_,
            is_shutdown());
        {
            AtomicHolder h(this);
            if (requestShutdown_)
            {
                return;
            }
            requestShutdown_ = true;
            if (sleeping_)
            {
                sleeping_ = false;
                timer_.ensure_triggered();
            }
            if (isConnected_)
            {
                isConnected_ = false;
                notify();
            }
        }
        // NOTE: It would be nice to abort any pending asynchronous tasks we
        // have such as a connect or a getaddrinfo call running on the
        // secondary executors. However, there is no API to do that right now.
    }
 
    static int connect(const char *host, int port)
    {
        return connect(host, integer_to_string(port).c_str());
    }
 
    static int connect(const char *host, const char* port_str);
 
    static int connect(struct addrinfo *addr);
 
    static bool address_to_string(
        struct addrinfo *addr, string *host, int *port);
 
    static AddrinfoPtr string_to_address(const char *host, int port)
    {
        return string_to_address(host, integer_to_string(port).c_str());
    }
 
    static AddrinfoPtr string_to_address(
        const char *host, const char *port_str);
 
private:
    void prepare_strategy()
    {
        unsigned ofs = 0;
        auto search = params_->search_mode();
        // If we only have one extra thread, we initiate mdns only at the time
        // we are trying to connect to it. If we have two, we start the lookup
        // at the beginning.
        bool mdns_ahead = (connectExecutor_ != mdnsExecutor_);
        if (mdns_ahead && search != SocketClientParams::MANUAL_ONLY)
        {
            strategyConfig_[ofs++] = Attempt::INITIATE_MDNS;
        }
        if (params_->enable_last())
        {
            strategyConfig_[ofs++] = Attempt::RECONNECT;
        }
        switch (search)
        {
            case SocketClientParams::AUTO_MANUAL:
                if (!mdns_ahead)
                {
                    strategyConfig_[ofs++] = Attempt::INITIATE_MDNS;
                }
                strategyConfig_[ofs++] = Attempt::CONNECT_MDNS;
                strategyConfig_[ofs++] = Attempt::CONNECT_STATIC;
                break;
            case SocketClientParams::MANUAL_AUTO:
                strategyConfig_[ofs++] = Attempt::CONNECT_STATIC;
                if (!mdns_ahead)
                {
                    strategyConfig_[ofs++] = Attempt::INITIATE_MDNS;
                }
                strategyConfig_[ofs++] = Attempt::CONNECT_MDNS;
                break;
            case SocketClientParams::MANUAL_ONLY:
                strategyConfig_[ofs++] = Attempt::CONNECT_STATIC;
                break;
            case SocketClientParams::AUTO_ONLY:
                if (!mdns_ahead)
                {
                    strategyConfig_[ofs++] = Attempt::INITIATE_MDNS;
                }
                strategyConfig_[ofs++] = Attempt::CONNECT_MDNS;
                break;
        }
        if (params_->one_shot())
        {
            strategyConfig_[ofs++] = Attempt::FAILED_EXIT;
        }
        else
        {
            strategyConfig_[ofs++] = Attempt::WAIT_RETRY;
        }
        HASSERT(ofs <= strategyConfig_.size());
    }
 
    Action start_connection()
    {
        startTime_ = os_get_time_monotonic();
        prepare_strategy();
        {
            AtomicHolder h(this);
            strategyOffset_ = 0;
            mdnsPending_ = false;
            mdnsJoin_ = false;
            isConnected_ = false;
            mdnsAddr_.reset();
        }
        return call_immediately(STATE(next_step));
    }
 
    Action next_step()
    {
        Attempt a = Attempt::WAIT_RETRY;
        if (strategyOffset_ < strategyConfig_.size())
        {
            AtomicHolder h(this);
            if (requestShutdown_)
            {
                return exit();
            }
            a = strategyConfig_[strategyOffset_++];
        }
        switch (a)
        {
            default:
                DIE("Unexpected action");
            case Attempt::WAIT_RETRY:
                return wait_retry();
            case Attempt::FAILED_EXIT:
                return failed_oneshot();
            case Attempt::RECONNECT:
                return try_schedule_connect(SocketClientParams::CONNECT_RE,
                    params_->last_host_name(), params_->last_port());
            case Attempt::INITIATE_MDNS:
                return start_mdns();
            case Attempt::CONNECT_MDNS:
                return wait_and_connect_mdns();
            case Attempt::CONNECT_STATIC:
                return try_schedule_connect(SocketClientParams::CONNECT_MANUAL,
                    params_->manual_host_name(), params_->manual_port());
        }
    }
 
    Action try_schedule_connect(
        SocketClientParams::LogMessage log, string host, int port)
    {
        if (port <= 0 || host.empty())
        {
            return call_immediately(STATE(next_step));
        }
        string v = host;
        v += ':';
        v += integer_to_string(port);
        params_->log_message(log, v);
        fd_ = -1;
        n_.reset(this);
        connectExecutor_->add(new CallbackExecutable(
            [this, host, port]() { connect_blocking(host, port); }));
        return wait_and_call(STATE(connect_complete));
    }
 
    void connect_blocking(const string &host, int port)
    {
        AutoNotify an(&n_);
        auto addr = SocketClient::string_to_address(host.c_str(), port);
        if (params_->disallow_local() && local_test(addr.get()))
        {
            params_->log_message(SocketClientParams::CONNECT_FAILED_SELF);
            return;
        }
        fd_ = SocketClient::connect(addr.get());
        if (fd_ >= 0)
        {
            params_->set_last(host.c_str(), port);
        }
    }
 
    Action connect_complete()
    {
        if (fd_ < 0)
        {
            // reconnect failed
            return next_step();
        }
        else
        {
            // we have a connection.
            return connected();
        }
    }
 
    Action connected()
    {
        {
            AtomicHolder h(this);
            if (requestShutdown_)
            {
                ::close(fd_);
                fd_ = -1;
                return exit();
            }
            isConnected_ = true;
        }
        callback_(fd_, this);
        return wait_and_call(STATE(start_connection));
    }
 
    string to_string(const char *p)
    {
        if (!p || !*p)
        {
            return string();
        }
        return p;
    }
 
    Action start_mdns()
    {
        HASSERT(mdnsExecutor_);
        mdnsAddr_.reset();
        string srv = params_->mdns_service_name();
        string host = params_->mdns_host_name();
        if (!srv.empty())
        {
            {
                AtomicHolder h(this);
                mdnsPending_ = true;
            }
            mdnsExecutor_->add(new CallbackExecutable(
                [this, host, srv]() { mdns_lookup(host, srv); }));
        }
        return call_immediately(STATE(next_step));
    }
 
    void mdns_lookup(string mdns_hostname, string mdns_service)
    {
        int ai_ret = -1;
        struct addrinfo hints;
 
        memset(&hints, 0, sizeof(hints));
        hints.ai_family = AF_INET;
        hints.ai_socktype = SOCK_STREAM;
        hints.ai_flags = 0;
        hints.ai_protocol = IPPROTO_TCP;
 
        struct addrinfo *addr = nullptr;
        params_->log_message(SocketClientParams::MDNS_SEARCH, mdns_service);
        ai_ret = MDNS::lookup(mdns_service.c_str(), &hints, &addr);
        mdnsAddr_.reset(addr); // will take care of freeing it.
        if (ai_ret != 0 || addr == nullptr)
        {
            params_->log_message(SocketClientParams::MDNS_NOT_FOUND);
            // LOG(INFO, "mdns lookup for %s failed.", mdns_service.c_str());
        }
        else
        {
            params_->log_message(SocketClientParams::MDNS_FOUND);
        }
        {
            AtomicHolder h(this);
            mdnsPending_ = false;
            if (mdnsJoin_)
            {
                // Flow is waiting for mdns result.
                mdnsJoin_ = false;
                notify();
            }
        }
    }
 
    Action wait_and_connect_mdns()
    {
        {
            AtomicHolder h(this);
            if (mdnsPending_)
            {
                mdnsJoin_ = true;
                return wait_and_call(STATE(connect_mdns));
            }
        }
        return call_immediately(STATE(connect_mdns));
    }
 
    Action connect_mdns()
    {
        string host;
        int port = -1;
        if (!mdnsAddr_.get() ||
            !SocketClient::address_to_string(mdnsAddr_.get(), &host, &port))
        {
            // no address to connect to.
            return call_immediately(STATE(next_step));
        }
        return try_schedule_connect(
            SocketClientParams::CONNECT_MDNS, std::move(host), port);
    }
 
    Action failed_oneshot()
    {
        params_->log_message(SocketClientParams::CONNECT_FAILED_ONESHOT);
        return exit();
    }
    Action wait_retry()
    {
        {
            AtomicHolder h(this);
            if (requestShutdown_)
            {
                return exit();
            }
            sleeping_ = true;
        }
        long long end_time = startTime_ + SEC_TO_NSEC(params_->retry_seconds());
        timer_.start_absolute(end_time);
        return wait_and_call(STATE(sleep_done));
    }
 
    Action sleep_done()
    {
        {
            AtomicHolder h(this);
            sleeping_ = false;
        }
        return call_immediately(STATE(start_connection));
    }
 
    static bool local_test(struct addrinfo *addr);
 
    long long startTime_;
    StateFlowTimer timer_{this};
 
    std::unique_ptr<SocketClientParams> params_;
 
    std::function<void(int, Notifiable *)> callback_ = nullptr;
 
    ExecutorBase *connectExecutor_ = nullptr;
    ExecutorBase *mdnsExecutor_ = nullptr;
 
    std::array<Attempt, 5> strategyConfig_{{
        Attempt::WAIT_RETRY,
    }};
    uint8_t strategyOffset_ : 3;
 
    uint8_t mdnsPending_ : 1;
 
    uint8_t mdnsJoin_ : 1;
 
    uint8_t sleeping_ : 1;
 
    uint8_t requestShutdown_ : 1;
 
    uint8_t isConnected_ : 1;
 
    AddrinfoPtr mdnsAddr_;
 
    BarrierNotifiable n_;
 
    int fd_;
 
    DISALLOW_COPY_AND_ASSIGN(SocketClient);
};
 
#endif /* _UTILS_SOCKET_CLIENT_HXX */